package com.atguigu.gmall;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author caodan
 * @version 1.0
 * @date 2022-09-27 16:08
 * 案例演示：flinkSQL用法
 */
public class FlinkSQL {

    public static void main(String[] args) {

        //TODO 1.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(20));

        //TODO 2. 创建socket流读取指定端口数据 并进行类型转换
        SingleOutputStreamOperator<Emp> empDS = env.socketTextStream("10.100.2.78", 8888).map(
                lineStr -> {
                    String[] fieldArr = lineStr.split(",");
                    return new Emp(Integer.parseInt(fieldArr[0]), fieldArr[1], Integer.parseInt(fieldArr[2]), Long.parseLong(fieldArr[3]));
                }
        );

        SingleOutputStreamOperator<Dept> deptDS = env.socketTextStream("10.100.2.78", 9999).map(
                lineStr -> {
                    String[] fieldArr = lineStr.split(",");
                    return new Dept(Integer.parseInt(fieldArr[0]), fieldArr[1], Long.parseLong(fieldArr[2]));
                }
        );
        //TODO 3.将流转换为动态表
        tableEnv.createTemporaryView("emp",empDS);
        tableEnv.createTemporaryView("dept",deptDS);


//        tableEnv.executeSql("select e.empno,e.ename,d.deptno,d.dname from emp e join dept d on e.deptno = d.deptno").print();

        tableEnv.executeSql("select e.empno,e.ename,d.deptno,d.dname from emp e left join dept d on e.deptno = d.deptno").print();
    }
}
